package com.root.edu.realtime.util;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.checkerframework.checker.units.qual.C;

import java.util.Properties;

public class MyKafkaUtil {
    private static final String KAFKA_SERVER = "hadoop103:9092,hadoop104:9092,hadoop105:9092";
    //获取消费者对象
    public static final FlinkKafkaConsumer<String> getKafkaConsumer (String topic,String groupId) {
        Properties props = new Properties ();
        props.setProperty ( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,KAFKA_SERVER );
        props.setProperty ( ConsumerConfig.GROUP_ID_CONFIG,groupId );
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<> ( topic, new KafkaDeserializationSchema<String> () {
            @Override
            public boolean isEndOfStream(String s) {
                return false;
            }

            @Override
            public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
                if (consumerRecord != null && consumerRecord.value () != null) {
                    return new String ( consumerRecord.value () );
                }

                return null;
            }

            @Override
            public TypeInformation<String> getProducedType() {
                return TypeInformation.of ( String.class );
            }
        }, props );
        return kafkaConsumer;

    }























}
